package com.quanquan.user.message;


import com.alibaba.fastjson.JSON;
import com.quanquan.user.service.autoCompute.JavaSqlUtil;
import com.quanquan.user.service.autoCompute.SQLHandler;
import com.quanquan.user.service.autoCompute.constant.SynConst;
import com.quanquan.user.service.autoCompute.entity.GroupLabelRelation;
import com.quanquan.user.service.autoCompute.entity.LabelGroup;
import com.quanquan.user.service.autoCompute.entity.LabelGroupSynRecord;
import com.quanquan.user.service.autoCompute.entity.MemberLabel;
import com.quanquan.user.service.autoCompute.repository.GroupLabelRelationRepository;
import com.quanquan.user.service.autoCompute.repository.LabelGroupRepository;
import com.quanquan.user.service.autoCompute.repository.LabelGroupSynRecordRepository;
import com.quanquan.user.service.autoCompute.service.MemberLabelService;
import com.rabbitmq.client.Channel;
import exceptions.service.ServiceRuntimeException;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;
import utils.ReadAndWriteJson;
import utils.StringUtils;

import javax.annotation.Resource;
import javax.persistence.EntityManager;
import javax.persistence.PersistenceContext;
import javax.persistence.Query;
import javax.transaction.Transactional;
import java.io.IOException;
import java.math.BigInteger;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

@Slf4j
@Component
public class LabelSyncReceiver {

    @Resource
    private LabelGroupRepository labelGroupRepository;

    @Resource
    private GroupLabelRelationRepository groupLabelRelationRepository;

    @PersistenceContext
    private EntityManager entityManager;

    @Autowired
    private MemberLabelService memberLabelService;

    @Autowired
    private LabelGroupSynRecordRepository labelGroupSynRecordRepository;

    @Value("${user-label.path}")
    private String userLabelFilePath;

    /**
     * 异步更新标签
     */
    @RabbitListener(
            bindings = {
                    @QueueBinding(
                            value = @Queue(value = "syn_member_label", durable = "true"),
                            exchange = @Exchange(value = "memberLabel.exchange", type = "topic"),
                            ignoreDeclarationExceptions = "true",
                            key = "memberLabel.#"
                    )
            })
    @RabbitHandler
    @Transactional
    public void synLabel(Channel channel, @Payload String msg, @Headers Map<String, Object> headers) {
        Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
        try {
            channel.basicAck(deliveryTag, false);
        } catch (IOException e) {
            log.error("确认标签消息[{}]失败,原因：{}", msg, e.getMessage());
        }
        String labelId = (String) headers.get("memberLabelId");
        if (StringUtils.isEmpty(labelId)) {
            log.warn("标签ID不能为空");
            return;
        }
        MemberLabel label;
        try {
            label = memberLabelService.getById(labelId);
        } catch (ServiceRuntimeException e) {
            log.warn("未查询到该标签：{}", labelId);
            return;
        }
        label.setLastSyncTime(new Date());
        String countSql = SQLHandler.assembleLabelCount(label);
        log.info("同步标签[{}]sql语句：{}", label.getId(), countSql);
        try {
            Query queryCount = entityManager.createNativeQuery(countSql);
            BigInteger bi = (BigInteger) queryCount.getSingleResult();
            long i = bi.longValue();
            log.info("同步标签数据成功，执行结果：{}", i);
            label.setContainsNum(i);
            label.setSynStatus(SynConst.SUCCESS);
        } catch (Exception e) {
            log.warn("同步标签数据失败，原因：{}", e.getCause());
            label.setSynStatus(SynConst.FAILED);
        } finally {
            entityManager.close();
            memberLabelService.save(label);
        }
    }

    /**
     * 异步更新画像
     */
    @RabbitListener(
            bindings = {
                    @QueueBinding(
                            value = @Queue(value = "syn_label_group", durable = "true"),
                            exchange = @Exchange(value = "labelGroup.exchange", type = "topic"),
                            ignoreDeclarationExceptions = "true",
                            key = "labelGroup.#"
                    )
            })
    @RabbitHandler
    @Transactional
    public void synGroupLabel(Channel channel, @Payload String msg, @Headers Map<String, Object> headers) {

        log.info("同步用户画像消费者");
        Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
        try {
            channel.basicAck(deliveryTag, false);
        } catch (IOException e) {
            log.error("确认画像消息[{}]失败,原因：{}", msg, e.getCause());
        }
        String labelGroupId = (String) (headers.get("labelGroupId"));
        if (StringUtils.isEmpty(labelGroupId)) {
            log.warn("画像ID不能为空");
            return;
        }
        LabelGroup labelGroup = labelGroupRepository.findById(labelGroupId).orElse(null);
        if (null == labelGroup) {
            log.warn("未查询到该用户画像：{}", labelGroupId);
            return;
        }
        labelGroup.setLastSyncTime(new Date());
        List<GroupLabelRelation> relations = groupLabelRelationRepository.findByGroupId(labelGroupId);
        if (null == relations || relations.isEmpty()) {
            log.warn("未查询到该用户画像的子标签：{}", labelGroupId);
            return;
        }
        List<String> labelIds = relations.stream().map(relation -> relation.getLabelId()).collect(Collectors.toList());
        List<MemberLabel> labels = memberLabelService.getByIds(labelIds);
        if (labels == null || labels.isEmpty()) {
            log.warn("画像[{}]未包含标签", labelGroupId);
            return;
        }
        Map<String, List<MemberLabel>> map = memberLabelService.list2Map(labels);
        String countSql = JavaSqlUtil.assembleCountSql(map);
        log.info("画像[{}]查询数量sql语句：{}", labelGroup.getName(), countSql);
        try {
            Query count = entityManager.createNativeQuery(countSql);
            BigInteger cont = (BigInteger) count.getSingleResult();
            long i = cont.longValue();
            log.info("画像[{}]数据同步成功，计数结果：{}条", labelGroup.getName(), i);
            labelGroup.setContainsNum(i);
            labelGroup.setSynStatus(SynConst.SUCCESS);
        } catch (Exception e) {
            log.warn("画像[{}]数据同步失败，原因：{}", labelGroup.getName(), e.getCause());
            labelGroup.setSynStatus(SynConst.FAILED);
        } finally {
            entityManager.close();
            labelGroupRepository.save(labelGroup);
        }
    }

    /**
     * 异步更新画像
     */
    @RabbitListener(
            bindings = {
                    @QueueBinding(
                            value = @Queue(value = "syn_label_group_user", durable = "true"),
                            exchange = @Exchange(value = "labelGroup.exchange", type = "topic"),
                            ignoreDeclarationExceptions = "true",
                            key = "syncUser.#"
                    )
            })
    @RabbitHandler
    @Transactional
    public void synGroupLabelUser(Channel channel, @Payload String msg, @Headers Map<String, Object> headers) {

        log.info("同步用户画像数据快照");
        log.info("channel:{}",JSON.toJSONString(channel));
        log.info("msg:{}",msg);
        Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
        try {
            channel.basicAck(deliveryTag, false);
        } catch (IOException e) {
            log.error("同步用户画像数据快照[{}]失败,原因：{}", msg, e.getCause());
        }

        String labelGroupId = (String) (headers.get("labelGroupId"));
        if (StringUtils.isEmpty(labelGroupId)) {
            log.warn("画像ID不能为空");
            return;
        }
        LabelGroup labelGroup = labelGroupRepository.findById(labelGroupId).orElse(null);
        if (null == labelGroup) {
            log.warn("未查询到该用户画像：{}", labelGroupId);
            return;
        }

        System.out.println(JSON.toJSONString(headers));

        Long synRecordId = Long.parseLong(headers.get("synRecordId").toString());
        if (null == synRecordId) {
            log.warn("同步记录ID不能为空");
            return;
        }
        LabelGroupSynRecord synRecord = labelGroupSynRecordRepository.findById(synRecordId).orElse(null);
        if (null == synRecord) {
            log.warn("未查询到同步记录：{}", synRecordId);
            return;
        }
        synRecord.setUpdateTime(new Date());
        List<GroupLabelRelation> relations = groupLabelRelationRepository.findByGroupId(labelGroupId);
        if (null == relations || relations.isEmpty()) {
            log.warn("未查询到该用户画像的子标签：{}", labelGroupId);
            return;
        }
        List<String> labelIds = relations.stream().map(relation -> relation.getLabelId()).collect(Collectors.toList());
        List<MemberLabel> labels = memberLabelService.getByIds(labelIds);
        if (labels == null || labels.isEmpty()) {
            log.warn("画像[{}]未包含标签", labelGroupId);
            return;
        }
        Map<String, List<MemberLabel>> map = memberLabelService.list2Map(labels);
        String selectSql = JavaSqlUtil.assembleSelectSql(map);
        log.info("画像[{}]查询数量sql语句：{}", labelGroup.getName(), selectSql);
        try {
            Query select = entityManager.createNativeQuery(selectSql);

            List resultList = select.getResultList();

            ReadAndWriteJson.writeFile(userLabelFilePath + "synRecord_" + labelGroupId + "_" + synRecordId + ".json",JSON.toJSONString(resultList));

            synRecord.setSynStatus(SynConst.SUCCESS);
            synRecord.setContainsNum(resultList.size());

        } catch (Exception e) {
            log.warn("画像[{}]数据同步失败，原因：{}", labelGroup.getName(), e.getCause());
            synRecord.setSynStatus(SynConst.FAILED);
        } finally {
            entityManager.close();
            labelGroupSynRecordRepository.save(synRecord);
        }
    }

}
